package com.cz.general;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.TimeUnit;

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("master:9876");
        // 异步发送失败后不进行重试
        producer.setRetryTimesWhenSendAsyncFailed(0);
        // 指定新创建的Topic的Queue数量为2
        producer.setDefaultTopicQueueNums(2);
        producer.start();

        for (int i = 0; i < 10; i++) {
            byte[] body = ("Hi" + i).getBytes();
            try {
                Message message = new Message("myTopicA", "myTag", body);
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        // 由于采用的是异步发送，所以若不sleep，则消息还未发送就会将producer给关闭
        TimeUnit.SECONDS.sleep(3);
        producer.shutdown();

    }
}


